package com.feishi.core.util.queue.cxq;



import com.feishi.core.util.queue.Customer;
import com.feishi.core.util.queue.Producer;
import com.feishi.core.util.queue.feishi.Msg;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: caixq
 * Date: 2018-09-05
 * Time: 下午4:06
 */
public class CxqProducer<T,R> implements Producer<Msg<T,R>> {
    private ExecutorService pool;
    private Customer<Msg<T,R>> customer;
    private Map<T,CxqExecutor<T,R>> executorMap= new HashMap<>();

    @Override
    public void push(Msg<T, R> msg) {
        T key = msg.getGroup();
        CxqExecutor executor=executorMap.get(key);
        if(executor==null) {
            executor = new CxqExecutor<T,R>(customer,pool);
            executorMap.put(key,executor);
        }

        executor.push(msg);
        executor.execute();
    }

    @Override
    public void close() {

    }

    public CxqProducer(Customer<Msg<T,R>> customer) {
        this.customer = customer;
        this.pool= Executors.newCachedThreadPool();
    }

    public CxqProducer( Customer<Msg<T, R>> customer,ExecutorService pool) {
        this.pool = pool;
        this.customer = customer;
    }
}
